import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;

/**
 * @author Tang
 * @classname NoContext
 * @description [  ]
 * @date 2021/5/9 9:57
 */
public class NoContext {

    public void test01() {


    }


    @Test
    public void test() throws Exception {
        Mono.defer(() -> {
            return Mono.just(1);
        }).then(Mono.just(2)).subscribe(System.out::println);

        Flux.push((fluxSink -> {
            Flux.range(1, 100).map(fluxSink::next).subscribe();
            fluxSink.onCancel(() -> {
                System.out.println("取消shi...");
            });
            fluxSink.onDispose(() -> {
                System.out.println("invoked for complete, error, or cancel signals....");
            });
            fluxSink.complete();
            fluxSink.onRequest(System.out::println);
        })).limitRate(10).buffer(10)
                .subscribe(System.out::println);

        Mono<String> mono = Mono.just("hello ");
        Thread thread = new Thread(() -> {
            mono.map(msg -> msg + "Thread ")
                    .subscribe(v -> System.out.println(v + Thread.currentThread().getName()));
        });
        thread.start();
        thread.join();

        Flux.interval(Duration.ofMillis(300), Schedulers.single())
                .subscribe(v -> {
                    System.out.println(Thread.currentThread().getName() + "--->" + v);
                });

        Schedulers.immediate().schedule(() -> {
            System.out.println(Thread.currentThread().getName() + "--->");
        });

        Schedulers.single().schedule(() -> {
            System.out.println(Thread.currentThread().getName() + "--->");
        });

        Schedulers.boundedElastic();
    }

    public static void main(String[] args) throws Exception {
        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);
        Flux<String> flux = Flux.range(1, 100)
                .map(i -> 10 + i)
                .subscribeOn(s)
                .map(i -> Thread.currentThread().getName() + "value " + i);
        new Thread(() -> flux.subscribe(System.out::println, (ex) -> {
        }, s::dispose)).start();

    }
}